package com.javacoo.event.client.internal.guava;

import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.springframework.core.task.TaskExecutor;

import com.google.common.eventbus.AsyncEventBus;
import com.google.common.eventbus.DeadEvent;
import com.google.common.eventbus.EventBus;
import com.google.common.eventbus.Subscribe;
import com.javacoo.event.client.api.EventHelper;

import lombok.extern.slf4j.Slf4j;

/**
 * EventHelper接口实现类
 * <p>说明:</p>
 * <li>基于google eventbus</li>
 * @author duanyong@jccfc.com
 * @date 2021/10/15 17:38
 */
@Slf4j
public class EventBusEventHelper implements EventHelper {
    /**
     * EventBus
     */
    private static EventBus eventBus = null;
    /**
     * AsyncEventBus
     */
    private static AsyncEventBus asyncEventBus = null;
    /**
     * DelayQueue
     */
    private static DelayQueue<EventItem> delayQueue = null;
    /**
     * 延迟时间执行器
     */
    private static TaskExecutor delayTaskExecutor = null;

    public EventBusEventHelper() {
        delayTaskExecutor = new TaskExecutor() {
            ExecutorService executorService = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), Runtime.getRuntime().availableProcessors(),
                    0L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<>(), Executors.defaultThreadFactory());
            @Override
            public void execute(Runnable task) {
                executorService.execute(task);
            }
        };
        eventBus = new EventBus();
        asyncEventBus = new AsyncEventBus(delayTaskExecutor);
        delayQueue = new DelayQueue<>();
        //注册DeadEvent监听
        eventBus.register(new Object() {
            @Subscribe
            public void lister(DeadEvent event) {
                log.error("{}发布的{}事件未找到监听对象", event.getSource().getClass(), event.getEvent());
            }
        });
        asyncEventBus.register(new Object() {
            @Subscribe
            public void lister(DeadEvent event) {
                log.error("{}发布的{}事件未找到监听对象", event.getSource().getClass(), event.getEvent());
            }
        });
    }
    /**
     * 发布同步事件
     * <li></li>
     *
     * @param eventObject : 事件对象
     * @author duanyong@jccfc.com
     * @date 2021/10/15 11:18
     * @return: void
     */
    @Override
    public void post(Object eventObject) {
        log.info("发布同步事件:{}",eventObject);
        eventBus.post(eventObject);
    }

    /**
     * 发布异步事件
     * <li></li>
     *
     * @param eventObject :事件对象
     * @author duanyong@jccfc.com
     * @date 2021/10/15 11:19
     * @return: void
     */
    @Override
    public void postAsync(Object eventObject) {
        log.info("发布异步事件:{}",eventObject);
        asyncEventBus.post(eventObject);
    }

    /**
     * 发布延迟时间
     * <li></li>
     *
     * @param eventObject : 事件对象
     * @param delayTime : 延迟时间,单位:毫秒
     * @author duanyong@jccfc.com
     * @date 2021/10/15 11:20
     * @return: void
     */
    @Override
    public void postDelay(Object eventObject, long delayTime) {
        log.info("延迟执行事件->入延迟队列:{}",eventObject);
        //入延迟队列
        delayQueue.put(new EventBusEventHelper.EventItem(eventObject, delayTime));
        //执行
        delayTaskExecutor.execute(()->execute());
    }

    /**
     * 注册监听对象
     * <p>说明:</p>
     * <li></li>
     *
     * @param listenerObject : 监听对象
     * @author duanyong@jccfc.com
     * @date 2021/10/15 18:01
     */
    @Override
    public void register(Object listenerObject) {
        log.info("注册监听对象:{}",listenerObject);
        asyncEventBus.register(listenerObject);
        eventBus.register(listenerObject);
    }

    /**
     * 注销监听对象
     * <p>说明:</p>
     * <li></li>
     *
     * @param listenerObject : 事件元数据
     * @author duanyong@jccfc.com
     * @date 2021/10/15 18:02
     */
    @Override
    public void unRegister(Object listenerObject) {
        log.info("注销监听对象:{}",listenerObject.toString());
        asyncEventBus.unregister(listenerObject);
        eventBus.unregister(listenerObject);
    }

    /**
     * 异步执行
     * <li></li>
     * @author duanyong@jccfc.com
     * @date 2021/7/8 10:08
     * @return: void
     */
    private void execute(){
        try {
            // 使用DelayQueue的take方法获取当前队列里的元素(take方法是阻塞方法，如果队列里有值则取出，否则一直阻塞)
            asyncEventBus.post(delayQueue.take().getEventObject());
            log.info("延迟执行事件");
        }catch (InterruptedException interruptedException){
            log.info("延迟执行事件异常:",interruptedException);
        }
    }

    /**
     * 事件项
     * <li></li>
     * @author duanyong@jccfc.com
     * @date 2021/7/8 9:06
     */
    class EventItem<T> implements Delayed {
        /** 触发时间:单位 毫秒 */
        private long time;
        /** 事件对象 */
        private T eventObject;

        public EventItem(T eventObject, long time) {
            super();
            // 将传入的时间转换为超时的时刻
            this.time = TimeUnit.NANOSECONDS.convert(time, TimeUnit.MILLISECONDS)
                    + System.nanoTime();
            this.eventObject = eventObject;
        }

        public long getTime() {
            return time;
        }

        public T getEventObject() {
            return eventObject;
        }

        @Override
        public long getDelay(TimeUnit unit) {
            // 剩余时间= 到期时间-当前系统时间，系统一般是纳秒级的，所以这里做一次转换
            return unit.convert(time- System.nanoTime(), TimeUnit.NANOSECONDS);
        }

        @Override
        public int compareTo(Delayed o) {
            // 剩余时间-当前传入的时间= 实际剩余时间（单位纳秒）
            long d = getDelay(TimeUnit.NANOSECONDS) - o.getDelay(TimeUnit.NANOSECONDS);
            // 根据剩余时间判断等于0 返回1 不等于0
            // 有可能大于0 有可能小于0  大于0返回1  小于返回-1
            return (d == 0) ? 0 : ((d > 0) ? 1 : -1);
        }

        @Override
        public String toString() {
            return "EventItem{" +
                    "time=" + time +
                    ", eventObject='" + eventObject + '\'' +
                    '}';
        }
    }
}
